﻿#include "DAMessageQueueProxy.h"
#include <QList>
#include <QMutex>
#include <QMutexLocker>
#include <QElapsedTimer>
#include <QApplication>
namespace DA
{
class DAMessageQueueProxyPrivate
{
    DA_IMPL_PUBLIC(DAMessageQueueProxy)
public:
    enum EmitSignalType
    {
        SignalQueueAppended    = 0,
        SignalQueueSizeChanged = 1
    };
    DAMessageQueueProxyPrivate(DAMessageQueueProxy* p);
    ~DAMessageQueueProxyPrivate();
    //_DAThreadSafeMessageQueue通知DAGlobalMessageQueuePrivate，队列有东西插入
    void queueAppended();

    //_DAThreadSafeMessageQueue通知DAGlobalMessageQueuePrivate，队列的尺寸变化了
    void queueSizeChanged();

public:
    QElapsedTimer _lastSignalEmitElapsedTimer;  ///< 记录最近一次发射信号的间隔
    int _emitIntervalms;                        ///发射间隔
};

/**
 * @brief 一个线程安全的队列，所有的@sa DAGlobalMessageQueue 都会向这个队列注册
 *
 * 这个使用单例，避免全局变量的初始化顺序问题导致异常
 */
class _DAThreadSafeMessageQueue
{
private:
    _DAThreadSafeMessageQueue() : _capacity(1000)
    {
    }

public:
    static _DAThreadSafeMessageQueue& getInstance()
    {
        static _DAThreadSafeMessageQueue s_queue;
        return s_queue;
    }

    int size() const
    {
        QMutexLocker lc(&_mutex);
        return _messages.size();
    }

    DAMessageLogItem at(int index) const
    {
        QMutexLocker lc(&_mutex);
        return _messages.value(index);
    }

    void append(const DAMessageLogItem& item)
    {
        bool needNotifySizeChanged = false;

        {
            QMutexLocker lc(&_mutex);
            if (_messages.size() >= _capacity) {
                _messages.pop_front();
            } else {
                //通知队列的尺寸变化了
                needNotifySizeChanged = true;
            }
            _messages.append(item);
        }
        //通知队列有东西插入了
        {
            QMutexLocker lc(&_mutexNotifys);
            if (needNotifySizeChanged) {
                for (DAMessageQueueProxyPrivate* p : _notifys) {
                    p->queueSizeChanged();
                }
            }
            for (DAMessageQueueProxyPrivate* p : _notifys) {
                p->queueAppended();
            }
        }
    }

    void setCapacity(int c)
    {
        QMutexLocker lc(&_mutex);
        _capacity = c;
    }

    int getCapacity()
    {
        QMutexLocker lc(&_mutex);
        return _capacity;
    }

    void registerNotify(DAMessageQueueProxyPrivate* p)
    {
        QMutexLocker lc(&_mutexNotifys);
        _notifys.append(p);
    }

    void unregisterNotify(DAMessageQueueProxyPrivate* p)
    {
        QMutexLocker lc(&_mutexNotifys);
        _notifys.removeAll(p);
    }

private:
    mutable QMutex _mutex;
    mutable QMutex _mutexNotifys;
    QList< DAMessageLogItem > _messages;            ///< 消息
    int _capacity;                                  ///< 容量
    QList< DAMessageQueueProxyPrivate* > _notifys;  ///< 等待通知的
};

}  // namespace DA

//===================================================
// using DA namespace -- 禁止在头文件using！！
//===================================================

using namespace DA;

//===================================================
// DAMessageQueueProxyPrivate
//===================================================
DAMessageQueueProxyPrivate::DAMessageQueueProxyPrivate(DAMessageQueueProxy* p) : q_ptr(p), _emitIntervalms(1000)
{
    _lastSignalEmitElapsedTimer.start();
    _DAThreadSafeMessageQueue::getInstance().registerNotify(this);
}

DAMessageQueueProxyPrivate::~DAMessageQueueProxyPrivate()
{
    _DAThreadSafeMessageQueue::getInstance().unregisterNotify(this);
}

void DAMessageQueueProxyPrivate::queueAppended()
{
    if (_lastSignalEmitElapsedTimer.elapsed() > _emitIntervalms) {
        //需要通知到gui
        _lastSignalEmitElapsedTimer.restart();
        q_ptr->emitSignal(SignalQueueAppended);
    }
}

void DAMessageQueueProxyPrivate::queueSizeChanged()
{
    if (_lastSignalEmitElapsedTimer.elapsed() > _emitIntervalms) {
        //需要通知到gui
        _lastSignalEmitElapsedTimer.restart();
        q_ptr->emitSignal(SignalQueueSizeChanged);
    }
}
//===================================================
// DAMessageQueueProxy
//===================================================
DAMessageQueueProxy::DAMessageQueueProxy(QObject* par) : QObject(par), d_ptr(new DAMessageQueueProxyPrivate(this))
{
}

DAMessageQueueProxy::~DAMessageQueueProxy()
{
}

void DAMessageQueueProxy::append(const DAMessageLogItem& item)
{
    _DAThreadSafeMessageQueue::getInstance().append(item);
}

DAMessageLogItem DAMessageQueueProxy::at(int index) const
{
    return _DAThreadSafeMessageQueue::getInstance().at(index);
}

int DAMessageQueueProxy::size() const
{
    return _DAThreadSafeMessageQueue::getInstance().size();
}

/**
 * @brief 设置全局队列的容量
 * @param c
 */
void DAMessageQueueProxy::setGlobalQueueCapacity(int c)
{
    _DAThreadSafeMessageQueue::getInstance().setCapacity(c);
}

/**
 * @brief 获取全局队列的容量
 * @return
 */
int DAMessageQueueProxy::getGlobalQueueCapacity()
{
    return _DAThreadSafeMessageQueue::getInstance().getCapacity();
}

void DAMessageQueueProxy::emitSignal(int type)
{
    if (QApplication::startingUp() || QApplication::closingDown()) {
        //如果app还未启动,或者已经关闭不发射信号
        return;
    }
    switch (type) {
    case DAMessageQueueProxyPrivate::SignalQueueAppended:
        emit messageQueueAppended();
        break;
    case DAMessageQueueProxyPrivate::SignalQueueSizeChanged:
        emit messageQueueSizeChanged(_DAThreadSafeMessageQueue::getInstance().size());
        break;
    default:
        break;
    }
}
